package mq

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/admin"
	"github.com/apache/rocketmq-client-go/v2/consumer"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"github.com/apache/rocketmq-client-go/v2/producer"
	"wowCoupon.com/logger"
)

type Admin struct {
	Admin admin.Admin

	NameServerAddr []string
	BrokerAddr     string
	Topic          string
	Group          string
}

func NewAdmin(nameServerAddr []string, brokerAddr string, topic string, group string) *Admin {
	a := &Admin{
		NameServerAddr: nameServerAddr,
		BrokerAddr:     brokerAddr,
		Topic:          topic,
		Group:          group,
	}

	var err error
	a.Admin, err = admin.NewAdmin(
		admin.WithResolver(primitive.NewPassthroughResolver(a.NameServerAddr)),
	)

	if err != nil {
		fmt.Printf("create mq admin fail: %s\n", err)
		return nil
	}

	return a
}

func (a *Admin) CreateTopic(topic string) error {
	var err error

	//删除已存在的同名topic

	/*
		err = a.Admin.DeleteTopic(
			context.Background(),
			admin.WithTopicDelete(topic),
		)

		if err != nil {
			fmt.Printf("delete old topic fail: %s\n", err)
			//return err
		}*/

	//创建topic
	err = a.Admin.CreateTopic(
		context.Background(),
		admin.WithTopicCreate(topic),
		admin.WithBrokerAddrCreate(a.BrokerAddr),
		admin.WithWriteQueueNums(1),
		admin.WithReadQueueNums(1),
	)

	if err != nil {
		fmt.Printf("createTopic error: %s\n", err)
		return err
	}

	return nil
}

type Producer struct {
	Producer rocketmq.Producer
	Topic    string
}

func NewProducer(nameServerAddr []string, group string, topic string) *Producer {
	p := &Producer{
		Topic: topic,
	}

	var err error
	p.Producer, err = rocketmq.NewProducer(
		producer.WithNameServer(nameServerAddr),
		producer.WithRetry(2),
		producer.WithGroupName(group),
	)

	if err != nil {
		logger.Error("create mq producer fail: %s", err)
		return nil
	}

	return p
}

// 启动producer
func (p *Producer) Start() error {
	if err := p.Producer.Start(); err != nil {
		logger.Error("start producer error: %s", err.Error())
		return err
	}

	return nil
}

// 发送同步延时消息
func (p *Producer) SendSyncDelayMessage(message []byte, delayLevel int) error {
	// 创建消息
	msg := &primitive.Message{
		Topic: p.Topic,
		Body:  message,
		Queue: &primitive.MessageQueue{
			Topic:      p.Topic,
			BrokerName: "broker-a",
			QueueId:    0,
		},
	}

	//设置消息的tag，二级管理
	//msg.WithTag(mq.Tags)

	//设置消息延时等级，RocketMQ仅支持固定的几个级别的延迟时间
	msg.WithDelayTimeLevel(delayLevel)

	//发送消息
	logger.Debug("生产者发送延时消息")
	sendResult, err := p.Producer.SendSync(context.Background(), msg)

	if err != nil {
		logger.Error("send delay message error: %s", err)
		return err
	}

	if sendResult.Status != primitive.SendOK {
		err = errors.New("send delay msg to broker not ok")
		logger.Error(err.Error())
	}

	logger.Info("send delay message success: result=%s", sendResult.String())
	return nil
}

type Consumer struct {
	Consumer rocketmq.PushConsumer
	Topic    string
}

func NewConsumer(nameServerAddr []string, group string, topic string) *Consumer {
	c := &Consumer{
		Topic: topic,
	}

	var err error
	c.Consumer, err = rocketmq.NewPushConsumer(
		consumer.WithNameServer(nameServerAddr),
		//consumer.WithConsumerModel(consumer.Clustering),
		consumer.WithGroupName(group),
		//consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{"192.168.31.129:9876"})),
		consumer.WithConsumerOrder(true),
	)

	if err != nil {
		fmt.Printf("create mq consumer fail: %s\n", err)
		return nil
	}

	return c
}

func (c *Consumer) ConsumeMessage(callback func(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error)) error {
	var err error

	// 订阅并消费topic
	fmt.Println("订阅并消费topic")
	if err = c.Consumer.Subscribe(
		c.Topic,
		consumer.MessageSelector{},
		callback); err != nil {
		fmt.Printf("subscribe message error: %s\n", err)
		return err
	}

	// 启动consumer
	fmt.Println("启动consumer")
	if err = c.Consumer.Start(); err != nil {
		fmt.Printf("consumer start error: %s\n", err)
		return err
	}

	time.Sleep(10000 * time.Second)
	// 关闭consumer
	defer c.Consumer.Shutdown()

	return nil
}

func (c *Consumer) WithTags(tags string) {
	//c.Tags = tags
}
